package com.atguigu.education.app.dws;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.education.app.func.DimAsyncFunction;
import com.atguigu.education.bean.CourseReviewBean;
import com.atguigu.education.bean.TradeCourseBean;
import com.atguigu.education.util.ClickHouseUtil;
import com.atguigu.education.util.DateFormatUtil;
import com.atguigu.education.util.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.*;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * @title: DwsCourseReviewInfoWindow
 * @Author Mr.Liu
 * @Date: 2022/9/4 9:13
 * @Version 1.0
 */
public class DwsCourseReviewInfoWindow {
    public static void main(String[] args) throws Exception {
        // TODO 1 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // TODO 2 设置状态后端
        /*
        env.enableCheckpointing(5 * 60 * 1000L, CheckpointingMode.EXACTLY_ONCE );
        env.getCheckpointConfig().setCheckpointTimeout( 3 * 60 * 1000L );
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
        System.setProperty("HADOOP_USER_NAME", "atguigu");
         */
        String topic_name = "dwd_review_info";
        String groupID = "dws_course_review_info_window";
        DataStreamSource<String> reviewSource = env.addSource(KafkaUtil.getKafkaConsumer(topic_name,groupID));


        //order
        SingleOutputStreamOperator<JSONObject> jsonStream = reviewSource.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String s, Collector<JSONObject> collector) throws Exception {
                JSONObject jsonObject = JSONObject.parseObject(s);
                if (jsonObject.getString("review_stars") != null && "0".equals(jsonObject.getString("deleted"))) {
                    collector.collect(jsonObject);
                }
            }
        });

        SingleOutputStreamOperator<CourseReviewBean> flatMap = jsonStream.flatMap(new RichFlatMapFunction<JSONObject, CourseReviewBean>() {
            @Override
            public void flatMap(JSONObject jsonObject, Collector<CourseReviewBean> collector) throws Exception {
                Long goodCount = 0L;
                if ("5".equals(jsonObject.getString("review_stars"))) {
                    goodCount = 1L;
                }
                CourseReviewBean courseReviewBean = new CourseReviewBean("", "",jsonObject.getString("course_id"),"" ,1L, goodCount, Long.parseLong(jsonObject.getString("review_stars")), jsonObject.getLong("ts"));
                collector.collect(courseReviewBean);
            }
        });

        //水位线
        //filter.print("filter---");
        KeyedStream<CourseReviewBean, String> keyedStream = flatMap.keyBy(s -> s.getCourseId());

        SingleOutputStreamOperator<CourseReviewBean> beanWithWatermarkStream = keyedStream.assignTimestampsAndWatermarks(WatermarkStrategy.<CourseReviewBean>forBoundedOutOfOrderness(Duration.ofSeconds(2L)).withTimestampAssigner(new SerializableTimestampAssigner<CourseReviewBean>() {
            @Override
            public long extractTimestamp(CourseReviewBean courseReviewBean, long l) {
                return courseReviewBean.getTs()*1000L;
            }
        }));


        //开窗

       // beanWithWatermarkStream.print("water");

        SingleOutputStreamOperator<CourseReviewBean> reduce = beanWithWatermarkStream.windowAll(TumblingEventTimeWindows.of(Time.seconds(10L))).reduce(new ReduceFunction<CourseReviewBean>() {
            @Override
            public CourseReviewBean reduce(CourseReviewBean value1, CourseReviewBean value2) throws Exception {
                value1.setGoodCount(value1.getGoodCount() + value2.getGoodCount());
                value1.setReviewAmount(value1.getReviewAmount() + value2.getReviewAmount());
                value1.setReviewUserCount(value1.getReviewUserCount() + value2.getReviewUserCount());
                return value1;
            }
        }, new AllWindowFunction<CourseReviewBean, CourseReviewBean, TimeWindow>() {
            @Override
            public void apply(TimeWindow timeWindow, Iterable<CourseReviewBean> iterable, Collector<CourseReviewBean> collector) throws Exception {
                CourseReviewBean courseReviewBean = iterable.iterator().next();
                courseReviewBean.setStt(DateFormatUtil.toYmdHms(timeWindow.getStart()));
                courseReviewBean.setEdt(DateFormatUtil.toYmdHms(timeWindow.getEnd()));
                courseReviewBean.setTs(System.currentTimeMillis());
                collector.collect(courseReviewBean);
            }
        });
        SingleOutputStreamOperator<CourseReviewBean> courseBeanStream = AsyncDataStream.unorderedWait(reduce, new DimAsyncFunction<CourseReviewBean>("DIM_COURSE_FULL") {

            @Override
            public String getKey(CourseReviewBean input) {
                return input.getCourseId();
            }

            @Override
            public void join(CourseReviewBean input, JSONObject obj) {
                input.setCourseName(obj.getString("courseName"));
            }
        }, 100, TimeUnit.SECONDS);

        courseBeanStream.print("bean......");
        courseBeanStream.addSink(ClickHouseUtil.getClickHouseSinkFunc("insert into dws_course_review_suc_window values(?,?,?,?,?,?,?,?)"));
        env.execute(groupID);
    }
}
